package me.spring.cloud.common.components.rabbitmq.util;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;

/**
 * @description: mq消息工具类
 * @author: luffy
 * @create: 2018-12-14 11:37
 **/
public class RabbitMqUtil {
  public static String getDeadQueueName(String queneName){
    return queneName+"@dead";
  }


  /**
   * 创建延迟消息队列
   *
   * @param connectionFactory
   * @param queueName 队列名称
   * @param exchangeName exchange name
   * @param routingKey routingKey
   */
  public static void createDelayQueue(ConnectionFactory connectionFactory, String queueName, String exchangeName,
      String routingKey) throws IOException {
    Channel channel = connectionFactory.createConnection().createChannel(false);
    // 声明Exchange：主体，失败，重试
    createTopicExchange(channel, exchangeName);
    createTopicExchange(channel, exchangeName + "@failed");
    createTopicExchange(channel, exchangeName + "@retry");

    // 声明监听队列
    createQueue(channel, queueName);
    createQueue(channel, queueName + "@failed");
    createDelayQueue(channel, queueName + "@retry", exchangeName, queueName);

    // 绑定监听队列到Exchange
    queueBind(channel, queueName, exchangeName, routingKey);
    queueBind(channel, queueName, exchangeName, queueName);
    queueBind(channel, queueName + "@failed", exchangeName + "@failed", queueName);
    queueBind(channel, queueName + "@retry", exchangeName + "@retry", queueName);
  }

  /**
   * 创建Direct延迟消息队列
   *
   * @param connectionFactory
   * @param queueName 接受队列名称
   * @param exchangeName 接受exchange
   * @param routingKey 接受routingKey
   */
  public static void createDirectQueue(ConnectionFactory connectionFactory, String queueName, String exchangeName,
      String routingKey) throws IOException {
    Channel channel = connectionFactory.createConnection().createChannel(false);

    // 声明Exchange：主体，失败，重试
    createTopicExchange(channel, exchangeName);
    createTopicExchange(channel, exchangeName + "@failed");
    createTopicExchange(channel, exchangeName + "@retry");

    createQueue(channel, queueName);
    createDirectExchange(channel, exchangeName);
    queueBind(channel, queueName, exchangeName, routingKey);
  }


  /**
   * 发送普通消息
   *
   * @param amqpTemplate
   * @param routingKey 队列名称
   * @param message 消息
   */
  public static void sendMessage(AmqpTemplate amqpTemplate, String routingKey, Object message) {
    if (null != amqpTemplate && null != routingKey && null != message) {
      amqpTemplate.convertAndSend(routingKey, message);
    }
  }

  /**
   * @Return void
   * @Author luffy
   * @Description 发送订阅消息
   * @Date 2019-05-20 14:13
   * @Param [amqpTemplate, exchangeKey, message]
   **/
  public static void sendPSMessage(AmqpTemplate amqpTemplate, String exchangeKey, Object message) {
    if (null != amqpTemplate && null != exchangeKey && null != message) {
      amqpTemplate.convertAndSend(exchangeKey, "", message);
    }
  }

  public static void sendTopicMessage(AmqpTemplate amqpTemplate, String exchange, String routingKey, Object message) {
    if (null != amqpTemplate && null != exchange && null != message && null != routingKey) {
      amqpTemplate.convertAndSend(exchange, routingKey, message);
    }
  }

  /**
   * 发送延迟消费消息
   *
   * @param amqpTemplate
   * @param exchange exchange名称
   * @param routingKey 路由名称
   * @param message 消息
   * @param second 延迟时间（单位：秒）
   */
  public static void sendDelayMessage(AmqpTemplate amqpTemplate, String exchange, String routingKey, Object message,
      long second) {
    if (null != amqpTemplate && null != exchange && null != message && null != routingKey && second >= 0) {
      amqpTemplate.convertAndSend(exchange, routingKey, message, messagePostProcessor -> {
        messagePostProcessor.getMessageProperties().setExpiration(second * 1000 + "");
        return messagePostProcessor;
      });
    }
  }

  /**
   * 创建普通队列，并持久化
   *
   * @param channel
   * @param queue 队列名称
   */
  private static void createQueue(Channel channel, String queue) throws IOException {
    channel.queueDeclare(queue, true, false, false, null);
  }

  /**
   * 创建延迟消费队列
   *
   * @param channel
   * @param queueName 队列名称
   * @param exchange exchange名称
   * @param routingKey 转发消息key
   */
  private static void createDelayQueue(Channel channel, String queueName, String exchange, String routingKey)
      throws IOException {
    Map<String, Object> params = new HashMap<>(2);
    // x-dead-letter-exchange 声明了队列里的死信转发到的DLX名称，
    params.put("x-dead-letter-exchange", exchange);
    // x-dead-letter-routing-key 声明了这些死信在转发时携带的 routing-key 名称。
    params.put("x-dead-letter-routing-key", routingKey);
    channel.queueDeclare(queueName, true, false, false, params);
  }

  /**
   * 创建DirectExchange
   *
   * @param channel
   * @param exchange
   */
  private static void createDirectExchange(Channel channel, String exchange) throws IOException {
    channel.exchangeDeclare(exchange, BuiltinExchangeType.DIRECT, true);
  }

  /**
   * 创建TopicExchange
   *
   * @param channel
   * @param exchange
   */
  private static void createTopicExchange(Channel channel, String exchange) throws IOException {
    channel.exchangeDeclare(exchange, BuiltinExchangeType.TOPIC, true);
  }

  /**
   * 绑定queue exchange routingKey
   *
   * @param channel
   * @param queue
   * @param exchange
   * @param routingKey
   */
  private static void queueBind(Channel channel, String queue, String exchange, String routingKey)
      throws IOException {
    channel.queueBind(queue, exchange, routingKey);
  }


  /**
   * 删除队列
   *
   * @param connectionFactory
   * @param queue
   */
  public static void delQueue(ConnectionFactory connectionFactory, String queue) throws IOException {
    connectionFactory.createConnection().createChannel(false).queueDelete(queue);
  }

  /**
   * 组装消息配置
   *
   * @param ttl 消息有效时间 单位：毫秒
   */
  public static MessageProperties getMessageProperties(int ttl) {
    MessageProperties messageProperties = new MessageProperties();
    //设置消息的有效期 ms为单位
    messageProperties.setExpiration(ttl + "");
    return messageProperties;
  }
}
